import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.ArrayBlockingQueue;
public class Database2 {
    private Row rows[] = new Row[100];

    // thread pools for groups
    private Thread threadWorkers[] = new Thread[10];

    // lock for each group
    private Lock groupLocks[] = new Lock[10];

    // reentrant read write lock for each row
    // private ReentrantReadWriteLock[] readWriteLocks = new ReentrantReadWriteLock[100];

    // queue for intergroup producer-consumer pool
    private ArrayBlockingQueue<Transaction>[] producerConsumerQueues = new ArrayBlockingQueue[10];

    public Database2(){
        for (int i=0; i<100; i++){
            rows[i] = new Row(i);
        }
        // deprecate: for remove
        //for (int i=0; i<100; i++){
        //   readWriteLocks[i] = new ReentrantReadWriteLock();
        //}

        for (int i=0; i<10; i++){
            groupLocks[i] = new ReentrantLock();
        }

        for (int i=0; i<10; i++){
            producerConsumerQueues[i] = new ArrayBlockingQueue<Transaction>(100);
        }
    }
     

    public void executeTransactions(List<Transaction> transactions){
        for(int i = 0; i < 10; i++){
            // create runnable for each group
            int groupId = i;
            Thread thread = new Thread(() -> {
                groupRunnable(groupId);
            });
            threadWorkers[i] = thread;
            thread.start();
        }

        // pass in the transactions to the first queue
        for(Transaction t : transactions){
            // push the transaction to the first queue
            try {
                producerConsumerQueues[0].put(t);
            } catch (InterruptedException e){
                e.printStackTrace();
            }
        }

        for (Thread t : threadWorkers){
            // t.start();
            try {
                t.join();
            } catch (InterruptedException e){
                e.printStackTrace();
            }
        }
    }

    // runnable for each transaction, at each node
    public void groupRunnable(int groupId){
        // get transaction from the queue before it
        try {
            while (true) {
                Transaction transaction = producerConsumerQueues[groupId].take();
                // once detected the terminal transaction
                if (transaction.getId() == -1) {
                    // end of the transaction
                    if (groupId != 9) {
                        // end of the transaction
                        producerConsumerQueues[groupId + 1].put(transaction);
                        break;
                    }
                    break;
                }
                // traver the operations in the transaction
                for (Operation o : transaction.getOperations()) {
                    // TODO: if belongs to this group
                    if (o.getRowNumber() / 10 != groupId) {
                        // pass to the next
                    } else {
                        // read operation
                        if (o.getType() == 0) {
                            // apply reentrant read lock to the row
                            //readWriteLocks[o.getRowNumber()].readLock().lock();
                            o.setValue(rows[o.getRowNumber()].getValue());
                            // release the lock
                            //readWriteLocks[o.getRowNumber()].readLock().unlock();
                        } else {
                            // write operation
                            // apply reentrant write lock to the row
                            //readWriteLocks[o.getRowNumber()].writeLock().lock();
                            rows[o.getRowNumber()].setValue(o.getValue());
                            // release the lock
                            //readWriteLocks[o.getRowNumber()].writeLock().unlock();
                        }
                        System.out.println("Transaction" + transaction.getId() + "'s " + o);
                    }
                }
                // TODO: append to next group
                if (groupId != 9) producerConsumerQueues[(groupId + 1)].put(transaction);
            }
        }
        catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String []args){
        Transaction t1 = new Transaction(1);
        t1.addOperation(new Operation(0, 9, 0));
        t1.addOperation(new Operation(1, 54, 5));
        t1.addOperation(new Operation(0, 21, 0));

        Transaction t2 = new Transaction(2);
        t2.addOperation(new Operation(1, 9, 99));
        t2.addOperation(new Operation(0, 54, 0));
        t2.addOperation(new Operation(1, 21, 1234));

        Transaction t3 = new Transaction(3);
        t3.addOperation(new Operation(0, 1, 0));
        t3.addOperation(new Operation(0, 73, 0));
        t3.addOperation(new Operation(1, 54, 45));

        Transaction terminal = new Transaction(-1);

        LinkedList<Transaction> batch = new LinkedList<Transaction>();
        batch.add(t1);
        batch.add(t2);
        batch.add(t3);
        
        Database2 db = new Database2();
        Database dbSerial = new Database();

        // Generate all permutations
        List<List<Transaction>> allPermutations = Permutations.generate(new ArrayList<>(batch));
//        for (List<Transaction> transactions : allPermutations){
//            for (Transaction t : transactions){
//                System.out.print(t.getId() + " -> ");
//            }
//            System.out.println("END");
//        }

        // parallel execute all permutations
        // WARNING: do not remove. This is a placeholder to identify the end of feeding transactions
        batch.add(terminal);
        db.executeTransactions(batch);

        // serially execute all permutations, one by one
        for (List<Transaction> transactions : allPermutations){
            dbSerial.executeTransactions(transactions);

            boolean equivalent = true;
            for (int i=0; i<100; i++){
                if (db.rows[i].getValue() != dbSerial.getRows()[i].getValue()){
                    equivalent = false;
                    break;
                }
            }
            if (equivalent){
                System.out.print("This execution is equivalent to a serial execution of ");
                for (Transaction t : transactions) {
                    if (t.getId() == -1) continue;
                    System.out.print("Transaction " + t.getId() + " -> ");
                }
                System.out.println("END");
            }
            // reset the serial database
            for (int i=0; i<100; i++){
                dbSerial.getRows()[i] = new Row(i);
            }
        }
    }
}
